package com.cssrc.wenjie.hbase.service.intf;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.cssrc.wenjie.hbase.service.HBaseDDL;

/**   
* @des 表DDL操作接口实现类
* @author wenjie   
* @date 2018年8月31日 下午7:08:08 
* @version V1.0  
*/
public class HBaseDDLImpl implements HBaseDDL {
    public static Logger log = Logger.getLogger(HBaseDDLImpl.class);
    
    private static Connection conn = null;
    static {
        Configuration conf = HBaseConfiguration.create();
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    @Override
    public void createTable(String tableName, String... families) {
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            for (String columnFamily : families) {
                tableDescriptor.addFamily(new HColumnDescriptor(columnFamily));
            }
            admin.createTable(tableDescriptor);
        } catch (Exception e) {
            log.error(tableName + "创建失败", e);
        }
        
    }
    
    @Override
    public void deleteTable(String tableName) {
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        } catch (Exception e) {
            log.error(tableName + "删除失败", e);
        }
        
    }
    
    @Override
    public List<TableName> listTables() {
        List<TableName> tables = new ArrayList<TableName>();
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            Collections.addAll(tables, admin.listTableNames());
        } catch (Exception e) {
            log.error("", e);
        }
        return tables;
    }
    
    @Override
    public boolean isTableExist(String tableName) {
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            if (admin.tableExists(tableName)) {
                return true;
            }
        } catch (Exception e) {
            log.error("", e);
        }
        return false;
    }
    
    @Override
    public List<String> queryTable() {
        // TODO
        return null;
    }
    
    @Override
    public void addFamilies(String tableName, String... families) {
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            if (!admin.tableExists(tableName)) {
                return;
            }
            for (String family : families) {
                admin.addColumn(tableName, new HColumnDescriptor(Bytes.toBytes(family)));
            }
            admin.flush(TableName.valueOf(tableName));
        } catch (Exception e) {
            log.error(tableName + "添加族失败", e);
        }
        
    }
    
    @Override
    public void deleteFamilies(String tableName, String... families) {
        try {
            HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
            if (!admin.tableExists(tableName)) {
                return;
            }
            for (String family : families) {
                admin.deleteColumn(tableName, family);
            }
            admin.flush(TableName.valueOf(tableName));
        } catch (Exception e) {
            log.error(tableName + "删除族失败", e);
        }
        
    }
    
    @Override
    public List<String> listFamilies(String tableName) {
        // TODO
        return null;
    }
    
}
